package com.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JDBCUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @author Archie
 * @date 2021-10-18 18:30
 * @description
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

	private Connection phoenixConn;
	private ValueState<Boolean> isFirstState;
	private Jedis redisClient;

	/**
	 * 每个并行度起始执行一次
	 * @param parameters
	 * @throws Exception
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		// 创建Phoenix连接
		/* 加载驱动 & 获取连接对象*/
		String url = Constant.PHOENIX_URL;
		phoenixConn = JDBCUtil.getPhoenixConnection(url);
		// 首条数据标记
		isFirstState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirstState", Boolean.class));
		redisClient = RedisUtil.getClient();
	}

	/**
	 * 流中每来一条数据，执行一次
	 * @param value
	 * @param context
	 * @throws Exception
	 */
	@Override
	public void invoke(Tuple2<JSONObject, TableProcess> value,
					   Context context) throws Exception {
		// 1. 检测目标表是否存在，不存在则建表
		checkAndCreateTable(value);
		// 2. 写入数据
		writeToPhoenix(value);
		// 3. 更新缓存
		updateCache(value);

	}

	private void updateCache(Tuple2<JSONObject, TableProcess> value) {

		JSONObject dim = value.f0;
		TableProcess tp = value.f1;

		String key = tp.getSinkTable() + ":" + dim.getLong("id");

		JSONObject upperDim = new JSONObject();
		if ("update".equals(tp.getOperateType()) && redisClient.exists(key)) {
			// 去更新缓存: 缓存存在才需要更新
			// 把dim中所有的字段名全部大写
			for (Map.Entry<String, Object> entry : dim.entrySet()) {
				upperDim.put(entry.getKey().toUpperCase(), entry.getValue());
			}
			redisClient.set(key, upperDim.toJSONString());
		}

		// 粗暴
		//redisClient.del(key);
	}

	/**
	 * 向表中写入数据
	 * @param value
	 */
	private void writeToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {
		JSONObject data = value.f0;
		TableProcess tp = value.f1;

		// 1. 拼接SQL --> e.g. upsert into T(...) values(?,?,...)
		StringBuilder sql = new StringBuilder();
		sql
				.append("upsert into ")
				.append(tp.getSinkTable() + "(")
				.append(tp.getSinkColumns())
				.append(")values(")
				.append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
				.append(")");

		System.out.println("=====> 插入数据到 " + tp.getSinkTable() + "！");
		System.out.println("-----> 建表语句为 " + sql.toString());

		// 2. 获取PrepareStatement
		PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
		/* 给SQL中的占位符赋值 */
		String[] columnNames = tp.getSinkColumns().split(",");
		for (int i = 0; i < columnNames.length; i++) {
			Object targetValue = data.get(columnNames[i]); // targetValue 可能为 Null
			/*
				下面不能用 targetValue + "" ！
				---------------------------------------------
				String s = null;
				System.out.println(s); // 本质是 null
				System.out.println(s instanceof String); // false
				System.out.println((s + "")); // 本质是 "null"
				System.out.println((s + "") instanceof String); // true
				---------------------------------------------
				所以要主动判定
			*/
			ps.setString(i + 1, targetValue == null ? null : targetValue.toString());
		}

		// 3. 执行 & 关闭
		ps.execute();
		phoenixConn.commit();
		ps.close();

	}

	/**
	 * 检查并创建表（仅第一条数据过来时候才建表）
	 * @param value
	 */
	private void checkAndCreateTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
		if (isFirstState.value() == null) { /*  首次数据标记*/
			System.err.println("=====> 建表语句执行！");
			TableProcess tp = value.f1;
			StringBuilder sql = new StringBuilder();
			sql
					.append("create table if not exists ")
					.append(tp.getSinkTable())
					.append("(")
					.append(tp.getSinkColumns().replaceAll(",", " varchar,"))
					.append(" varchar, constraint pk primary key(")
					.append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
					.append("))")
					.append(tp.getSinkExtend() == null ? "" : tp.getSinkExtend()); // 加盐，预分区->指定分区个数

			PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
			ps.execute();
			phoenixConn.commit();
			ps.close();

			isFirstState.update(true);
		}
	}

	@Override
	public void close() throws Exception {
		if (phoenixConn != null) {
			phoenixConn.close();
		}
	}
}
